package cn.nkpro.elcube.mq;

import cn.nkpro.elcube.cloud.CloudConstants;
import cn.nkpro.elcube.cloud.NkCloudProperties;
import cn.nkpro.elcube.cloud.NkCloudSyncEvent;
import cn.nkpro.elcube.cloud.NkCloudSyncPayload;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * ELCubeCloud 云内数据同步配置
 *
 * 作用如下：
 * 1、注册云同步需要的 MQ交换机及队列
 * 2、监听云同步队列，转换成SpringEvent，推送给各个相关模块
 *
 */
@Slf4j
@ConditionalOnProperty("nk.cloud.id")
@Configuration
public class NkCloudSyncMQListenerConfig {

    @Autowired
    private NkCloudProperties cloudProperties;

    /**
     * 初始化消息消费者监听器
     */
    @Bean
    public NkCloudChannelAwareMessageListener NkCloudChannelAwareMessageListener(){
        return new NkCloudChannelAwareMessageListener();
    }
    /**
     * 定义消费者监听器
     */
    public static class NkCloudChannelAwareMessageListener implements ChannelAwareMessageListener,
            ApplicationEventPublisherAware {

        private ApplicationEventPublisher applicationEventPublisher;

        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            try{
                String json = new String(message.getBody());
                NkCloudSyncPayload payload = JSON.parseObject(json, NkCloudSyncPayload.class);
                applicationEventPublisher.publishEvent(new NkCloudSyncEvent(this,payload.getProducerId(), payload.getType(),payload.getPayload()));
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }catch (Exception e){
                log.error("云同步发生错误："+e.getMessage(),e);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            }
        }

        @Override
        public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
            this.applicationEventPublisher = applicationEventPublisher;
        }
    }

    /**
     * 创建一个MQ监听容器，监听 elcube.cloud.[cloudId]队列
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory, NkCloudChannelAwareMessageListener handle){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);
        container.setQueueNames(CloudConstants.QUEUE_NAME + cloudProperties.getId());
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setExposeListenerChannel(true);
        container.setMessageListener(handle);
        return container;
    }
}
